package qdu.edu.com.fushanf4.service.sort.topn;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import qdu.edu.com.fushanf4.service.sort.IntWritableSortReducer;

import java.io.IOException;

public class IntWritableSortAndTopNReducer extends IntWritableSortReducer {
    private int count = 0;
    private static int topN = 0;                // topN

    public static void setTopN(int topN) {
        IntWritableSortAndTopNReducer.topN = topN;
    }

    @Override
    protected void setup(Context context) {
        if (null != context.getConfiguration().get("topN")) {    // 方便单独测试map，一种权衡
            setTopN(Integer.parseInt(context.getConfiguration().get("topN")));
        }
    }

    /**
     * 该类取前1500，可以并列，即如果有3个key的值都是1498，那么这三个都会保存到文件中，即结果会有1501个
     * 该类重写了IntWritableSortReducer的reduce方法，用来在sort的基础上，取TopN
     */
    @Override
    protected void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        if (count < topN) {
            for (Text keyOut : values) {
                context.write(keyOut, key);
                count++;
            }
        }
    }
}
